[AURON #2378] Support runtime filters in native Iceberg scan#2379
[AURON #2378] Support runtime filters in native Iceberg scan#2379lyne7-sc wants to merge 11 commits into
Conversation
|
Thanks @weiqingy for the review. I addressed the concrete issues and kept the shim boundary intentionally. PTAL |
|
Thanks for the quick turnaround. The concrete fixes read well: letting |
There was a problem hiding this comment.
Pull request overview
Adds support for Spark runtime filters (used by dynamic partition pruning) to Auron’s native Iceberg scan path by carrying BatchScanExec.runtimeFilters into native execution and re-planning Iceberg partitions with those filters applied at runtime, plus integration coverage for common formats and changelog scans.
Changes:
- Thread runtime filters into
NativeIcebergTableScanExecand re-plan Iceberg scan tasks/partitions with runtime filters before native execution. - Cache runtime-filtered Iceberg scan plans separately from static scan plans and add a Spark-version shim for copying
BatchScanExecwith runtime filters. - Add Iceberg integration tests covering dynamic partition pruning for Parquet/ORC, empty-pruning, and changelog scans.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala | Adds integration tests and helpers to assert native Iceberg scan sees runtime filters and DPP reduces files/partitions (including empty-pruning and changelog join). |
| thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeIcebergTableScanExec.scala | Carries runtime filters into the native scan node, re-plans Iceberg scan with runtime-filtered partitions, and canonicalizes including runtime filters. |
| thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala | Adds runtime-filter-aware plan caching and reflective extraction of runtime-filtered partitions (filteredPartitions) when available. |
| thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergConvertProvider.scala | Passes BatchScanExec.runtimeFilters into NativeIcebergTableScanExec. |
| spark-extension/src/main/scala/org/apache/spark/sql/auron/Shims.scala | Introduces shim API to copy BatchScanExec while preserving runtime filters across Spark versions. |
| spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/auron/ShimsImpl.scala | Implements copyBatchScanExecWithRuntimeFilters for Spark 3.0–4.1 constructor signature differences. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| private def runtimeFilteredPartitions(exec: BatchScanExec): Option[Seq[InputPartition]] = { | ||
| if (exec.runtimeFilters.isEmpty) { | ||
| return None | ||
| } | ||
|
|
||
| exec.prepare() | ||
| MethodUtils.invokeMethod(exec, true, "waitForSubqueries") | ||
| invokeDeclaredMethod(exec, "filteredPartitions") match { | ||
| case Some(seq: scala.collection.Seq[_]) => | ||
| Some(flattenPartitions(seq)) | ||
| case _ => | ||
| None | ||
| } | ||
| } |
|
@lyne7-sc, please address the review comments of Copilot. |
Which issue does this PR close?
Closes #2378
Rationale for this change
Spark's
BatchScanExeccan carry runtime filters for dynamic partition pruning. These filters are resolved after planning and can reduce the input partitions/files that a scan should read.Auron's native Iceberg scan should preserve and apply these runtime filters so dynamic partition pruning can reduce Iceberg scan work on the native path.
What changes are included in this PR?
BatchScanExec.runtimeFiltersintoNativeIcebergTableScanExec.BatchScanExecwith runtime filters across supported Spark versions.Are there any user-facing changes?
Queries using native Iceberg scan can read fewer partitions/files when Spark dynamic partition pruning applies. No API change.
How was this patch tested?
Added
AuronIcebergIntegrationSuitecases.